package com.atguigu.mapreduce.reducejoin;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class ReduceJoinMapper extends Mapper<LongWritable, Text, Text, OrderPd> {
    private Text k = new Text();
    private OrderPd v = new OrderPd();
    String fileName;

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, OrderPd>.Context context) throws IOException, InterruptedException {
        FileSplit inputSplit = (FileSplit) context.getInputSplit();
        fileName = inputSplit.getPath().getName();
    }

    /**
     * 数据整合
     *
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, OrderPd>.Context context) throws IOException, InterruptedException {
        String lineValue = value.toString();
        String[] split = lineValue.split("\t");
        // 对文件区分
        // 获取切片对象
        if (fileName.startsWith("order.txt")) {
            // 来自order文件
            k.set(split[1]);
            v.setOrderId(Integer.parseInt(split[0]));
            v.setpId(Integer.parseInt(split[1]));
            v.setAmount(Integer.parseInt(split[2]));
            v.setpName("");
            v.setTitle(fileName); // order
        } else {
            // 来自pd文件
            // 来自order文件
            k.set(split[0]);
            v.setOrderId(0);
            v.setpId(Integer.parseInt(split[0]));
            v.setAmount(0);
            v.setpName(split[1]);
            v.setTitle(fileName); // order
        }
        context.write(k, v);
    }
}
